package graph;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Vector;

import mapreduce.DynamicNumberFormat;
import mapreduce.FlipMapper;
import mapreduce.IntegerFormat;
import mapreduce.JobConf;
import mapreduce.Mapper;
import mapreduce.MultiTask;
import mapreduce.NullFormat;
import mapreduce.Record;
import mapreduce.RecordFeeder;
import mapreduce.RecordFormatWriter;
import mapreduce.RecordReader;
import mapreduce.RecordWriter;
import mapreduce.Reducer;
import mapreduce.StringFormat;
import mapreduce.Task;
import mapreduce.Record.RecordPositionType;

public class AssignIdTask extends MultiTask {
	static public File getIDFile(String idName) {
		return new File(idName + ".id");
	}

	public static AssignIdTask getTask(int taskId, File dataFile,
			String firstName, String secondName, File outFile) {
		AssignIdTask task = new AssignIdTask(null, taskId);
		File task1OutFile = new File("1.out");
		Task idTask1 = FirstIdTask.getTask(taskId * 1000 + 1, dataFile, 1,
				firstName, task1OutFile);
		task.addTask(idTask1);
		Task idTask2 = SecondIdTask.getTask(taskId * 1000 + 2, task1OutFile, 2,
				secondName, outFile);
		task.addTask(idTask2);
		return task;
	}

	private AssignIdTask(JobConf conf, int id) {
		super(conf, id);
	}

	/**
	 * (name1, name2 : ) map: -> (name1 : name2) reducer: -> (id1: name2)+
	 * 
	 * @author jinghe
	 * 
	 */
	static class FirstIdTask extends Task {
		static public FirstIdTask getTask(int taskId, File dataFile,
				int columnNum, String name, File outFile) {
			RecordFeeder feeder = new PairRelationFeeder(dataFile);
			RecordWriter writer = new RecordWriter();
			writer.open(outFile);
			Mapper mapper = new FirstIdTaskMapper();
			Reducer reducer = new FirstIdTaskReducer(getIDFile(name));
			JobConf conf = new JobConf(feeder, writer, mapper, reducer);
			return new FirstIdTask(conf, taskId);
		}

		private FirstIdTask(JobConf conf, int taskId) {
			super(conf, taskId);
		}
	}

	static class FirstIdTaskMapper implements Mapper {
		@Override
		public void map(Record record, Vector<Record> outRecords) {
			String[] values = (String[]) record.getKey().getData();
			Record.RecordData key = new Record.RecordData(values[0],
					new StringFormat());
			Record.RecordData value = new Record.RecordData(values[1],
					new StringFormat());
			outRecords.add(Record.getRecord(key, value));
		}

		@Override
		public void release(Vector<Record> outRecords) {
		}
	}

	static class FirstIdTaskReducer implements Reducer {

		private PrintWriter m_writer;

		public FirstIdTaskReducer(File idFile) {
			try {
				m_writer = new PrintWriter(new FileWriter(idFile));
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		int m_id = 0;

		@Override
		public void reduce(Vector<Record> inRecords, Vector<Record> outRecords) {
			String name = (String) inRecords.get(0).getKey().getData();
			m_writer.printf("%s %d\n", name, m_id);

			for (Record record : inRecords) {
				Record.RecordData newKey = new Record.RecordData(new Integer(
						m_id), new IntegerFormat());
				Record.RecordData newValue = new Record.RecordData(record.getValue().getData(), new StringFormat());
				outRecords.add(Record.getRecord(newKey, newValue));
			}

			m_id++;
		}

		@Override
		public void release(Vector<Record> outRecords) {
			m_writer.flush();
			m_writer.close();
		}
	}

	/**
	 * (id1: name2)
	 * 
	 * map: -> (name2: id1) reduce: -> (id1, id2:)
	 * 
	 * @author jinghe
	 * 
	 */
	static class SecondIdTask extends Task {
		static public SecondIdTask getTask(int taskId, File inFile,
				int columnNum, String name, File outFile) {
			RecordFeeder feeder = new RecordReader(inFile);
			RecordWriter writer = new RecordWriter();
			writer.open(outFile);
			Mapper mapper = new SecondIdTaskMapper();
			Reducer reducer = new SecondIdTaskReducer(getIDFile(name));
			JobConf conf = new JobConf(feeder, writer, mapper, reducer);
			return new SecondIdTask(conf, taskId);
		}

		private SecondIdTask(JobConf conf, int taskId) {
			super(conf, taskId);
		}
	}

	static class SecondIdTaskMapper implements Mapper {
		private FlipMapper m_mapper;

		public SecondIdTaskMapper() {
			Record.RecordPosition[] keyPositions = new Record.RecordPosition[1];
			keyPositions[0] = new Record.RecordPosition(RecordPositionType
					.getValueType(), 0);
			Record.RecordPosition[] valuePositions = new Record.RecordPosition[1];
			valuePositions[0] = new Record.RecordPosition(RecordPositionType
					.getKeyType(), 0);
			m_mapper = new FlipMapper(keyPositions, valuePositions);
		}

		@Override
		public void map(Record record, Vector<Record> outRecords) {
			m_mapper.map(record, outRecords);
		}

		@Override
		public void release(Vector<Record> outRecords) {
			m_mapper.release(outRecords);
		}
	}

	static class SecondIdTaskReducer implements Reducer {

		private PrintWriter m_idWriter;
		private int m_id = 0;

		public SecondIdTaskReducer(File idFile) {
			try {
				m_idWriter = new PrintWriter(new FileWriter(idFile));
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		@Override
		public void reduce(Vector<Record> inRecords, Vector<Record> outRecords) {
			String name = (String) inRecords.get(0).getKey().getData();
			m_idWriter.printf("%s %d\n", name, m_id);

			for (Record record : inRecords) {
				Integer[] keyData = new Integer[2];
				keyData[0] = (Integer) record.getValue().getData();
				keyData[1] = m_id;
				Record.RecordData newKey = new Record.RecordData(keyData,
						new DynamicNumberFormat(new IntegerFormat()));
				Record.RecordData newValue = new Record.RecordData(null, new NullFormat());
				outRecords.add(Record.getRecord(newKey, newValue));
			}

			m_id++;
		}

		@Override
		public void release(Vector<Record> outRecords) {
			m_idWriter.close();

		}

	}

	static public void main(String[] args) {
		File edgesDir = new File("../test-data/favorite.test.graph");
		File dataFile = new File("../test-data/favorite.test.dat");
		File outFile = new File("../test-data/favorite.idassign.dat");
		EdgesConfig config = new EdgesConfig("user", "tweets", "favorite");
		Task assignIdTask = AssignIdTask.getTask(1, dataFile,
				config.m_firstName, config.m_secondName, outFile);
		assignIdTask.run();
		IdManager firstIdManager = IdManager.getIdManager(edgesDir.toString(),
				config.m_firstName);
		firstIdManager.load(getIDFile(config.m_firstName));
		System.out.printf("%d %s\n", firstIdManager.getId("user2"),
				firstIdManager.getText(2));
		IdManager secondIdManager = IdManager.getIdManager(edgesDir.toString(),
				config.m_secondName);
		secondIdManager.load(getIDFile(config.m_secondName));
		System.out.printf("%d %s\n", secondIdManager.getId("5"),
				secondIdManager.getText(2));
	}
}
